package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: gaojian
 * @Description: 针对企查查t_eci_seriousviolation表和中数t_break_law表进行数据映射开发
 * @Date: 2021/12/29
 */

// ****************** TYPE字段暂时未处理 *****************

public class TEciSeriousviolation {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_eci_seriousviolation").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_break_law");
        long zs_count = zs_data.count();
        if(zs_count>0){

            // 处理步骤说明 1.UPDATE 2.DELETE
            // 1.UPDATE 这里修改id
            Dataset<Row> update_data = spark.sql("SELECT \n" +
                    "md5(CONCAT(\n" +
                    "UPPER(TRIM(A.CREDITCODE)),DATE_FORMAT(IF(A.INDATE = '' OR A.INDATE IS NULL,'1900-01-01',A.INDATE),'yyyy-MM-dd'),\n" +
                    "UPPER(TRIM(A.INREGORG)),DATE_FORMAT(IF(A.OUTDATE = '' OR A.OUTDATE IS NULL,'1900-01-01',A.OUTDATE),'yyyy-MM-dd') )) AS ID,\n" +
                    "B.KEY_NO,\n" +
                    "B.COMPANY_ID,\n" +
                    "B.COMPANY_NAME,\n" +
                    "'严重违法失信企业名单' AS TYPE,\n" +
                    "A.INDATE AS ADD_DATE,\n" +
                    "A.INREGORG AS ADD_OFFICE,\n" +
                    "A.INREASON AS ADD_REASON,\n" +
                    "A.OUTDATE AS REMOVE_DATE,\n" +
                    "A.OUTREGORG AS REMOVE_OFFICE,\n" +
                    "A.OUTREASON AS REMOVE_REASON,\n" +
                    "A.JOBID, '0' AS ISADD \n" +
                    "FROM \n" +
                    "(\n" +
                    "    select del.creditcode,l.* from a_dc_ep_ods.zs_t_break_law l\n" +
                    "    inner join a_dc_ep_ods.zs_t_break_law_del del on upper(trim(del.entid)) = upper(trim(l.entid))\n" +
                    "    where del.creditcode is not null and del.creditcode != ''\n" +
                    ") A INNER JOIN \n" +
                    "a_dc_ep_ods.t_eci_company B \n" +
                    "ON UPPER(TRIM(B.CREDIT_CODE)) = UPPER(TRIM(A.CREDITCODE))");
            update_data.createOrReplaceTempView("tmp_t_eci_seriousviolation");
            spark.sql("insert overwrite table a_dc_ep_incr.t_eci_seriousviolation select * from tmp_t_eci_seriousviolation");

            //2.DELETE
            spark.sql("select d.*,c.credit_code from a_dc_ep_dwi.t_eci_seriousviolation d inner join a_dc_ep_ods.t_eci_company c on d.company_id = c.company_id")
                    .createOrReplaceTempView("tmp_table7");
            Dataset<Row> delete_data = spark.sql("select \n" +
                    "s.id,\n" +
                    "s.key_no,\n" +
                    "s.company_id,\n" +
                    "s.company_name,\n" +
                    "s.type,\n" +
                    "s.add_date,\n" +
                    "s.add_office,\n" +
                    "s.add_reason,\n" +
                    "s.remove_date,\n" +
                    "s.remove_office,\n" +
                    "s.remove_reason,\n" +
                    "del.jobid,'-1' as isadd from \n" +
                    "( select del.creditcode,del.jobid from a_dc_ep_ods.zs_t_break_law_del del \n" +
                    "left join a_dc_ep_ods.zs_t_break_law l on upper(trim(l.entid)) = upper(trim(del.entid))\n" +
                    "where l.entid is null  and del.creditcode != '' and del.creditcode is not null) del\n" +
                    "inner join tmp_table7 s on upper(trim(s.credit_code)) = upper(trim(del.creditcode))");
            delete_data.createOrReplaceTempView("tmp_t_eci_seriousviolation2");
            spark.sql("insert into table a_dc_ep_incr.t_eci_seriousviolation select * from tmp_t_eci_seriousviolation2");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_eci_seriousviolation");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
